package com.flute.haflute.agent;

import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.manage.JvmId;
import com.flute.haflute.agent.manage.JvmManager;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterConstants.CLI;
import com.flute.haflute.common.ClusterConstants.COMMAND;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.slave.Child;

/**
 * 以一项工作执行为集合的SLAVE角色节点的管理，主要是Child JVM的启动，监控等
 * 
 * @see JobControlCenterManager
 */
public class JobSlaveManager {
	private static Logger logger = LoggerFactory
			.getLogger(JobSlaveManager.class);
	private static boolean debugMore = true;
	private static String currentMaster;

	/**
	 * 启动执行任务的Child JVM（子JVM）
	 * 
	 * @param master
	 *            该任务的工作控制节点
	 */
	public static void startTaskExecutorJvm(String master) {
		File libPath = new File("./lib");
		JvmManager jvmManager = JvmManager.getInstance(Child.class.getName(),
				libPath);

		// should check if has child, if has shutdown first. actually currently,
		// jobtracker will exit when task finished
		if (jvmManager.getAllProcess().size() > 0) {
			syncJvmStatusWitchManager(jvmManager, master);
			// shutdownAllChildJvms(jvmManager);
		}

		if (jvmManager.getAllProcess().size() == 0) {
			startAllChildJvms(jvmManager, master);
		}

		/*
		 * old strategy which will not end the child jvm int startedProessNum =
		 * jvmManager.getAllProcess().size(); if (startedProessNum == 0) { // 启动
		 * slave.Child startAllChildJvms(); } else { //notify master changed
		 * logger.info("master {} changed to {} ",currentMaster, master);
		 * JvmManager.sendCommand2AllChilds(COMMAND.JOB_NEW_CENTER + master);
		 * currentMaster = master; }
		 */
	}

	private static void syncJvmStatusWitchManager(JvmManager jvmManager,
			String master) {
		Map<JvmId, Process> processes = jvmManager.processMap();
		for (Iterator<JvmId> it = processes.keySet().iterator(); it.hasNext();) {
			JvmId id = it.next();
			Process proc = jvmManager.getProcess(id);
			try {
				int exitCode = proc.exitValue();
				logger.info("one process exit code={}", exitCode);
				it.remove();
			} catch (IllegalThreadStateException e) {
			}

			logger.info("master {} changed to {} ", currentMaster, master);
			JvmManager.sendCommand2AllChilds(COMMAND.JOB_NEW_CENTER + master);
			currentMaster = master;
		}
	}

	private static void shutdownAllChildJvms(JvmManager jvmManager) {
		Collection<Process> processes = jvmManager.getAllProcess();
		logger.info("total {} child-processes to finish", processes.size());
		for (Process process : processes) {
			JvmManager.sendCommand2OneJvm(process, COMMAND.SHUTDOWN);
			logger.info("wait for one Child process to finish");
			try {
				process.waitFor();
			} catch (InterruptedException e) {
			}
			logger.info("one Child process Finished");
		}
		jvmManager.clearProcessCache();
	}

	private static void startAllChildJvms(JvmManager jvmManager, String master) {
		int numProcesses = getNumOfChilds();
		for (int i = 0; i < numProcesses; i++) {
			File workDirFile = new File("./slave" + i);
			if (!workDirFile.exists()) {
				if (!workDirFile.mkdirs()) {
					System.err.println("can not Make workdir: master/");
				}
			}
			startNewChild(jvmManager, workDirFile, master);
			logger.info("started one child in directory: {} ",
					workDirFile.getName());
		} // for
	}

	private static int getNumOfChilds() {
		return ClusterContext.getCpucores() / 4 + 1;
	}

	private static void startNewChild(JvmManager jvmManager, File workDirFile,
			String master) {
		ArrayList<String> argList = new ArrayList<String>();
		argList.add(CLI.JOB_MASTER_IS + master);
		argList.add(CLI.ROOTDIR_IS + ClusterContext.getRootDir());
		String[] jvmParams = new String[0];
		if (ClusterContext.isJvmDebugging() && debugMore) {
			jvmParams = new String[2];
			jvmParams[0] = "-Xdebug";
			jvmParams[1] = "-Xrunjdwp:transport=dt_socket,address=6668,server=y,suspend=n";
			debugMore = false; // only debug one jvm otherwise it's a big task
		}

		try {
			Process process = jvmManager.getProcess(jvmManager.spawnChildJvm(
					workDirFile, jvmParams,
					argList.toArray(new String[argList.size()])));
			BufferedReader reader = new BufferedReader(new InputStreamReader(
					process.getInputStream()));
			String line;
			while ((line = reader.readLine()) != null) {
				// 重要！该项控制必须一致
				if (ClusterConstants.PROCESS_STARTED.trim().equals(line)) {
					break;
				}
				System.out.println(line);
			} // while
			reader.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void main(String args[]) {
		JobSlaveManager.startTaskExecutorJvm("127.0.0.1");
		System.exit(0);
	}
}
